package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDim;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import com.atguigu.edu.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection hBaseConnection;
    //private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();
        //jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConnection);
        //RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        JSONObject jsonObject = value.f0;
        TableProcessDim tableProcessDim = value.f1;
        //根据type确定对Hbase做何种操作
        String type = jsonObject.getString("type");
        jsonObject.remove("type");
        //id,activity_name,activity_type,activity_desc,start_time,end_time,create_time
        String sinkTableName = tableProcessDim.getSinkTable();
        String rowkey = jsonObject.getString(tableProcessDim.getSinkRowKey());
        String sinkFamily = tableProcessDim.getSinkFamily();

        if("delete".equals(type)){
            HBaseUtil.deleteRow(hBaseConnection, Constant.HBASE_NAMESPACE,sinkTableName,rowkey);
        } else {
            HBaseUtil.putRow(hBaseConnection,Constant.HBASE_NAMESPACE,sinkTableName,rowkey,sinkFamily,jsonObject);
        }
/*        if("delete".equals(type)||"update".equals(type)){
            System.out.println("从redis中清除"+sinkTableName+"表的"+rowkey+"缓存数据");
            jedis.del(RedisUtil.getKey(sinkTableName,rowkey));
        }*/
    }
}
